package com.shujia.spark.kafka

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo4SparkOnKafkaDriect {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setAppName("receiver")

    val sc = new SparkContext(conf)

    //创建sparkStreaming上下文对象，指定batch间隔时间
    val ssc = new StreamingContext(sc, Durations.seconds(5))


    /**
      * receiver模式底层使用的是kafka官方api,  消费偏移量会自动提交到zookeeper
      *
      * direct模式: 需要我们自己管理偏移量， 可以将偏移量保存的hbase中
      * 自己管理偏移量：我们可以在数据处理完成之后更新消费偏移量，
      *
      */

    val kafkaParams = Map[String, String](
      "metadata.broker.list" -> "node1:9092"
      // "auto.offset.reset" -> "largest"
    )

    // direct模式： topic中一个分区对应rdd中一个分区
    val topics = Set("flume")

    /* val directDS: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
       ssc,
       kafkaParams,
       topics
     )*/

    // 体检处理kafka返回的数据， 里面一般只需要取出value就可以
    val messageHandlerFun = (messageHandler: MessageAndMetadata[String, String]) => {
      // 取出value
      messageHandler.message()
    }

    val groupId = "asdasasdwasd"

    val topic = "flume"

    /**
      * 3、创建表
      */
    OffsetManagerUtil.createTable(groupId)

    //指定初始读取数据的偏移量,  每一个topic的分区都需要指定
    //去外部数据库中获取偏移量
    /**
      * 2、获取初始偏移量
      */
    var fromOffsets = OffsetManagerUtil.getOffSet(groupId)
    println("获取到的偏移量：" + fromOffsets)
    if (fromOffsets == null) {
      fromOffsets = Map(TopicAndPartition(topic, 0) -> 0L)
    }

    val directDS: InputDStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
      ssc,
      kafkaParams,
      fromOffsets,
      messageHandlerFun
    )

    directDS.foreachRDD(rdd => {

      val count = rdd.count()
      println("处理数据逻辑,  处理数据量为：" + count)

      /**
        * 当计算逻辑完成之后，将偏移量保存到外部数据库中，比如hbase
        *
        */

      //获取当前处理数据的偏移量
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      offsetRanges.foreach(offsetRange => {

        //当前batch开始的偏移量
        val fromOffset = offsetRange.fromOffset
        //当前batch结束的偏移量
        val untilOffset = offsetRange.untilOffset

        val topic = offsetRange.topic

        val partition = offsetRange.partition

        println(s"$fromOffset\t$untilOffset\t$topic\4$partition")

        /**
          * 将偏移量保存到外部数据库
          *
          * 1、更新偏移量
          */
        OffsetManagerUtil.updateOffSet(groupId, topic, partition, untilOffset)

      })
    })

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
